package com.liuyz.essync.utils;

import com.liuyz.essync.entity.*;
import io.searchbox.client.JestClient;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.JestResult;
import io.searchbox.client.config.HttpClientConfig;
import io.searchbox.core.Index;
import io.searchbox.core.Search;
import io.searchbox.core.SearchScroll;
import io.searchbox.params.Parameters;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.util.*;

@Component
public class EsUtil {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    Es1 es1;

    @Autowired
    Es2 es2;

    @Autowired
    SyncConfigBean syncConfigBean;

    @Autowired
    JdbcTemplate jdbcTemplate;

    // ES1
    private JestClient jestClient1;

    // ES2
    private JestClient jestClient2;

    @PostConstruct
    private void init() {
        // ES1
        JestClientFactory factory1 = new JestClientFactory();
        HttpClientConfig.Builder builder1 = new HttpClientConfig.Builder("http://" + es1.getIp() + ":" + es1.getPort());
        builder1.connTimeout(80000)
                .readTimeout(80000)
                .multiThreaded(true);
        if (!StringUtils.isAnyBlank(es1.getUsername(), es1.getPassword())) {
            builder1.defaultCredentials(es1.getUsername(), es1.getPassword());
        }
        factory1.setHttpClientConfig(builder1.build());
        jestClient1 = factory1.getObject();
        if (null != jestClient1) {
            logger.info("---------------- ES1 对象加载完成！ ----------------");
        } else {
            throw new RuntimeException("---------------- ES1 对象加载异常！ ----------------");
        }

        // ES2
        JestClientFactory factory2 = new JestClientFactory();
        HttpClientConfig.Builder builder2 = new HttpClientConfig.Builder("http://" + es2.getIp() + ":" + es2.getPort());
        builder2.connTimeout(80000)
                .readTimeout(80000)
                .multiThreaded(true);
        if (!StringUtils.isAnyBlank(es2.getUsername(), es2.getPassword())) {
            builder2.defaultCredentials(es2.getUsername(), es2.getPassword());
        }
        factory2.setHttpClientConfig(builder2.build());
        jestClient2 = factory2.getObject();

        if (null != jestClient2) {
            logger.info("---------------- ES2 对象加载完成！ ----------------");
        } else {
            throw new RuntimeException("---------------- ES2 对象加载异常！ ----------------");
        }

        // 配置读取
        if (CollectionUtils.isEmpty(syncConfigBean.getSyncList())) {
            logger.info("---------------- 配置读取为空，禁止操作！ ----------------");
        }
    }

    // 同步数据方法
    public synchronized void sync() throws IOException {
        if (null == jestClient1) {
            throw new RuntimeException("ES1连接异常，禁止操作！");
        }
        if (null == jestClient2) {
            throw new RuntimeException("ES2连接异常，禁止操作！");
        }

        if (CollectionUtils.isEmpty(syncConfigBean.getSyncList())) {
            throw new RuntimeException("配置读取为空，禁止操作！");
        }

        // 开始同步
        logger.info("---------------- ES1往ES2 普通数据同步-开始！时间：" + new Date() + " ----------------");

        int count = 0;
        for (Sync sync : syncConfigBean.getSyncList()) {
            this.normalSync(sync);
            logger.info("---------------- 第【" + ++count + "】次同步：普通数据同步！ ----------------");
        }

        // 结束同步
        logger.info("---------------- ES1往ES2 普通数据同步-结束！时间：" + new Date() + " ----------------");
    }

    // 普通模式同步
    private void normalSync(Sync sync) throws IOException {
        // 参数完整性校验
        if (StringUtils.isNotBlank(sync.getOldIndexName()) && StringUtils.isNotBlank(sync.getOldType())
                && StringUtils.isNotBlank(sync.getNewIndexName()) && StringUtils.isNotBlank(sync.getNewType())) {

            // 获取ES1的数据（深度分页）
            Search search = new Search.Builder("{}").addIndex(sync.getOldIndexName()).addType(sync.getOldType())
                    .setParameter(Parameters.SIZE, 5000).setParameter(Parameters.SCROLL, "1m").build();
            JestResult jestResult = jestClient1.execute(search);
            List<String> searchResultList = jestResult.getSourceAsStringList();
            if (searchResultList.size() > 0) {
                for (String source : searchResultList) {
                    // 保存数据到ES2
                    jestClient2.execute(
                            new Index.Builder(source)
                                    .index(sync.getNewIndexName())
                                    .type(sync.getNewType())
                                    .build()
                    );
                }

                String scrollId = jestResult.getJsonObject().get("_scroll_id").getAsString();
                for (; ; ) {
                    SearchScroll scroll = new SearchScroll.Builder(scrollId, "1m").build();
                    JestResult result = jestClient1.execute(scroll);
                    List<String> list = result.getSourceAsStringList();
                    if (list.size() > 0) {
                        for (String source : list) {
                            // 保存数据到ES2
                            jestClient2.execute(
                                    new Index.Builder(source)
                                            .index(sync.getNewIndexName())
                                            .type(sync.getNewType())
                                            .build()
                            );
                        }
                    } else {
                        break;
                    }
                }
            }
        }
    }

    public void formSync(String sql, String keyWord, String ruler) throws IOException {
        if (null == jestClient1) {
            throw new RuntimeException("ES1连接异常，禁止操作！");
        }
        if (null == jestClient2) {
            throw new RuntimeException("ES2连接异常，禁止操作！");
        }

        if (CollectionUtils.isEmpty(syncConfigBean.getSyncList())) {
            throw new RuntimeException("配置读取为空，禁止操作！");
        }

        // 开始同步
        logger.info("---------------- ES1往ES2 规则数据同步-开始！时间：" + new Date() + " ----------------");

        // 获取查询参数
        List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
        List<String> keyList = new ArrayList<>();
        if (!CollectionUtils.isEmpty(list)) {
            for (Map<String, Object> map : list) {
                keyList.add(map.get("mapKey").toString());
            }
        }

        if (CollectionUtils.isEmpty(keyList)) {
            throw new RuntimeException("sql查询结果为空，不进行同步！");
        }

        int count = 0;
        for (Sync sync : syncConfigBean.getSyncList()) {
            // 组织查询参数并同步数据
            for (String key : keyList) {
                // 获取ES1的数据
                BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

                TypeQueryBuilder typeQueryBuilder = QueryBuilders.typeQuery(sync.getOldType());

                if (ruler.equals("equal")) {
                    MatchQueryBuilder matchQueryBuilder = new MatchQueryBuilder(keyWord, key);
                    boolQueryBuilder.must(matchQueryBuilder).must(typeQueryBuilder);
                } else if (ruler.equals("like")) {
                    QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery("*" + key + "*");
                    queryStringQueryBuilder.field(keyWord);
                    boolQueryBuilder.must(queryStringQueryBuilder).must(typeQueryBuilder);
                }

                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(boolQueryBuilder).size(10000);
                Search search = new Search.Builder(searchSourceBuilder.toString())
                        .addIndex(sync.getOldIndexName())
                        .build();
                List<String> sourceAsStringList = jestClient1.execute(search).getSourceAsStringList();

                // 保存数据到ES2
                if (!CollectionUtils.isEmpty(sourceAsStringList)) {
                    for (String source : sourceAsStringList) {
                        jestClient2.execute(
                                new Index.Builder(source)
                                        .index(sync.getNewIndexName())
                                        .type(sync.getNewType())
                                        .build());
                    }
                }
            }
            logger.info("---------------- 第【" + ++count + "】次同步：规则数据同步！ ----------------");
        }

        // 结束同步
        logger.info("---------------- ES1往ES2 规则数据同步-结束！时间：" + new Date() + " ----------------");
    }

    public List<String> formShow(String sql, String keyWord, String ruler) throws IOException {
        if (null == jestClient1) {
            throw new RuntimeException("ES1连接异常，禁止操作！");
        }
        if (null == jestClient2) {
            throw new RuntimeException("ES2连接异常，禁止操作！");
        }

        if (CollectionUtils.isEmpty(syncConfigBean.getSyncList())) {
            throw new RuntimeException("配置读取为空，禁止操作！");
        }

        // 获取查询参数
        List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);
        List<String> keyList = new ArrayList<>();
        if (!CollectionUtils.isEmpty(list)) {
            for (Map<String, Object> map : list) {
                keyList.add(map.get("mapKey").toString());
            }
        }

        if (CollectionUtils.isEmpty(keyList)) {
            throw new RuntimeException("sql查询结果为空，不进行同步！");
        }

        List<String> dataList = new ArrayList<>();

        for (Sync sync : syncConfigBean.getSyncList()) {
            // 组织查询参数并同步数据
            for (String key : keyList) {
                // 获取ES1的数据
                BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

                TypeQueryBuilder typeQueryBuilder = QueryBuilders.typeQuery(sync.getOldType());

                if (ruler.equals("equal")) {
                    MatchQueryBuilder matchQueryBuilder = new MatchQueryBuilder(keyWord, key);
                    boolQueryBuilder.must(matchQueryBuilder).must(typeQueryBuilder);
                } else if (ruler.equals("like")) {
                    QueryStringQueryBuilder queryStringQueryBuilder = QueryBuilders.queryStringQuery("*" + key + "*");
                    queryStringQueryBuilder.field(keyWord);
                    boolQueryBuilder.must(queryStringQueryBuilder).must(typeQueryBuilder);
                }

                SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
                searchSourceBuilder.query(boolQueryBuilder).size(10000);
                Search search = new Search.Builder(searchSourceBuilder.toString())
                        .addIndex(sync.getOldIndexName())
                        .build();
                List<String> sourceAsStringList = jestClient1.execute(search).getSourceAsStringList();

                if (!CollectionUtils.isEmpty(sourceAsStringList)) {
                    dataList.addAll(sourceAsStringList);
                }
            }
        }
        return dataList;
    }
}
